package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"fmt"
	"io/ioutil"
	"log"
	"math/rand"
	//	"bytes"
	"sync"
	"sync/atomic"
	"time"
	//	"6.824/labgob"
	"6.824/labrpc"
)

/*
	记录一下有关时间的一些约束
	Test要求：
	1. 在老leader挂掉之后，5s之内选出新leader
	2. 不同的peer使用不同的election timeout
	3. peer发送心跳的间隔不低于100ms

	election timeout是指在上次接收到心跳包之后，选举之前等待的时间，这个时间每个peer尽量不同，防止同时开始选举

	我选择的时间
	心跳间隔: 100ms
	超时间隔: 200ms
	election timeout：超时间隔 + 100~300ms
*/
const (
	heartbeatTimeout  time.Duration = 200 * time.Millisecond
	heartbeatInterval time.Duration = 100 * time.Millisecond
	none                            = -1
)

type RaftState int

const (
	Leader    RaftState = 0
	Candidate RaftState = 1
	Follower  RaftState = 2
)

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

//
// A Go object implementing a single Raft peer.
//
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

	// 下面是一些需要互斥访问的变量
	lock          sync.Mutex
	state         RaftState // 标志Raft的角色
	lastHeartBeat time.Time // 上次心跳时间
	leader        int       // 当前leader，没有则设置为-1
	currentTerm   int       // 当前term
	votedFor      int

	// Persistent state
	log []string

	// Volatile state on all servers
	commitIndex int
	lastApplied int

	// Volatile state on leaders
	nextIndex  []int
	matchIndex []int
}

func lock_guard(l *sync.Mutex, f func()) {
	l.Lock()
	defer l.Unlock()
	f()
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	// Your code here (2A).
	rf.lock.Lock()
	defer rf.lock.Unlock()
	return rf.currentTerm, rf.state == Leader
}

func RandomTime(start int, end int, unit time.Duration) time.Duration {
	return time.Duration(start+rand.Intn(end-start)) * unit
}

// Lab2A: Leader election
type AppendEntriesArgs struct {
	Term         int
	LeaderId     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []string
	LeaderCommit int
}

type AppendEntriesReply struct {
	Term    int
	Success bool
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	// 接收到了来自peer的日志记录请求，行为可能是：
	// 1. Entries为空，心跳包
	// 2. Entries不为空
	//fmt.Fprintf(os.Stderr, "%d received heartbeat: term[%d], leader[%d]\n", rf.me, args.Term, args.LeaderId)
	rf.lock.Lock()
	defer rf.lock.Unlock()

	rf.l(rf.currentTerm, fmt.Sprintf("received heartbeat: term[%d], leader[%d]\n", args.Term, args.LeaderId))
	reply.Term = rf.currentTerm
	if args.Term < rf.currentTerm {
		// 过时了
		reply.Success = false
		return
	}
	if args.Term > rf.currentTerm {
		// 当前Raft落后，退化为follower
		rf.state = Follower
		rf.currentTerm = args.Term
		rf.votedFor = none
		rf.lastHeartBeat = time.Now()
		rf.l(rf.currentTerm, "I am old, fall back")

		reply.Success = true
		return
	}
	// term相等
	if len(args.Entries) == 0 {
		// 心跳包
		rf.leader = args.LeaderId
		rf.state = Follower
		rf.lastHeartBeat = time.Now()

		reply.Success = true
		return
	}
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
}

//
// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
//
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).

	return true
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).

}

//
// example RequestVote RPC arguments structure.
// field names must start with capital letters!
//
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

//
// example RequestVote RPC reply structure.
// field names must start with capital letters!
//
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

//
// example RequestVote RPC handler.
//
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	// 有peer向当前Raft对象发送RequestVote, 要求当前Raft对象对其进行投票
	// 当前Raft对象在同一个term中只能给一个candidate投票
	// 不能投票的情况：
	// 1. 当前term中已经投过了
	// 2. 对方的term小于自己的term（对方滞后了）
	// 能投给对方的情况
	// 1. 对方的数据比当前的新
	rf.lock.Lock()
	defer rf.lock.Unlock()

	log.Printf("term[%d] #%d received requestVote, candidate[%d]\n", rf.currentTerm, rf.me, args.CandidateId)

	reply.VoteGranted = false

	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm
		log.Printf("term[%d] #%d don't vote for #%d: I am new\n", rf.currentTerm, rf.me, args.CandidateId)
		return
	}

	if args.Term > rf.currentTerm {
		// 对方比我新, 更新自身状态
		if args.Term > rf.currentTerm {
			rf.state = Follower
			rf.currentTerm = args.Term
			rf.votedFor = args.CandidateId
			reply.VoteGranted = true
			reply.Term = args.Term
			log.Printf("term[%d] #%d ===> #%d: you are new\n", rf.currentTerm, rf.me, args.CandidateId)
		} else if args.Term < rf.currentTerm {
			// 自身状态已修改
			reply.VoteGranted = false
			reply.Term = rf.currentTerm
			log.Printf("term[%d] #%d don't vote for #%d: I am new. changed after lock\n", rf.currentTerm, rf.me, args.CandidateId)
		} else {
			// eq
			if rf.votedFor == none {
				reply.VoteGranted = true
				reply.Term = rf.currentTerm
				rf.votedFor = args.CandidateId
				rf.state = Follower
				log.Printf("term[%d] #%d ===> #%d\n", rf.currentTerm, rf.me, args.CandidateId)
			} else {
				log.Printf("term[%d] #%d don't vote for #%d: Already voted\n", rf.currentTerm, rf.me, args.CandidateId)
			}
		}
		if reply.VoteGranted {
			log.Printf("term[%d] #%d vote for #%d\n", rf.currentTerm, rf.me, args.CandidateId)
		} else {
			log.Printf("term[%d] #%d don't vote for #%d\n", rf.currentTerm, rf.me, args.CandidateId)
		}
		return
	}
	// term相等
	if rf.votedFor != none {
		reply.Term = rf.currentTerm
		log.Printf("term[%d] #%d don't vote for #%d: term eq\n", reply.Term, rf.me, args.CandidateId)
		return
	}
	// 没投过票
	if rf.currentTerm == args.Term && rf.votedFor == none {
		reply.VoteGranted = true
		reply.Term = rf.currentTerm
		rf.votedFor = args.CandidateId
		log.Printf("term[%d] #%d ===> #%d\n", rf.currentTerm, rf.me, args.CandidateId)
		return
	} else {
		log.Printf("term[%d] #%d don't vote for #%d: already voted\n", rf.currentTerm, rf.me, args.CandidateId)
	}
}

//
// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
//
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	index := -1
	term := -1
	isLeader := true

	// Your code here (2B).

	return index, term, isLeader
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

// The ticker go routine starts a new election if this peer hasn't received
// heartsbeats recently.
func (rf *Raft) ticker() {
	for rf.killed() == false {

		// Your code here to check if a leader election should
		// be started and to randomize sleeping time using
		// time.Sleep().
		election_timeout := heartbeatTimeout + RandomTime(100, 300, time.Millisecond)
		time.Sleep(election_timeout)

		should_elect := false
		lock_guard(&rf.lock, func() {
			should_elect = (rf.leader != rf.me && time.Since(rf.lastHeartBeat) > heartbeatTimeout)
		})

		if should_elect {
			// 可以开始选举，等待一段时间开始选举，这个timeout可以防止多个peer同时开始选举
			// 开始一轮选举
			/*
				选举的过程中，需要考虑到自身的状态可能随时会发生变化，发生变化之后需要决定
				是否需要继续选举
			*/
			// 1. 增加自己的term
			rf.lock.Lock()
			rf.currentTerm++
			rf.votedFor = rf.me
			rf.state = Candidate
			//start_term := rf.currentTerm
			rf.l(rf.currentTerm, fmt.Sprintf("start election"))

			// timeout := false
			// 2. 向所有的server发送RequestVote
			args := RequestVoteArgs{
				Term:         rf.currentTerm,
				CandidateId:  rf.me,
				LastLogIndex: rf.lastApplied,
				LastLogTerm:  rf.lastApplied,
			}
			rf.lock.Unlock()

			// 创建多个协程，并发发送RPC请求，使用channel同步
			type RequestVoteReplyRet struct {
				Reply RequestVoteReply
				Ok    bool
			}
			res_channel := make(chan RequestVoteReplyRet, len(rf.peers)-1)
			for i := 0; i < len(rf.peers); i++ {
				/*
						每次拉票的时候，出现如下情况，选举结束
					1. 收到来自别的server表明自己是leader的消息

				*/
				if i != rf.me {
					t := i
					go func() {
						reply := RequestVoteReply{}
						//rf.l(rf.currentTerm, fmt.Sprintf("request #%d to vote", t))
						if rf.peers[t].Call("Raft.RequestVote", &args, &reply) {
							//rf.l(rf.currentTerm, fmt.Sprintf("got vote resp from #%d", t))
							res_channel <- RequestVoteReplyRet{Reply: reply, Ok: true}
						} else {
							//rf.l(rf.currentTerm, fmt.Sprintf("failed to request #%d", t))
							res_channel <- RequestVoteReplyRet{Reply: reply, Ok: false}
						}
					}()
				}
			}

			// request done
			// wait for response
			votes := 1
			win := false
			for i := 0; i < len(rf.peers)-1; i++ {
				replyRet := <-res_channel
				if replyRet.Ok {
					rf.lock.Lock()
					if rf.state != Candidate {
						log.Printf("term[%d] #%d state changed to %d during election. exiting", rf.currentTerm, rf.me, rf.state)
						rf.lock.Unlock()
						break
					}
					if replyRet.Reply.VoteGranted {
						votes++
					}
					// 如果当前所得票数大于半数，获胜
					if votes > len(rf.peers)/2 {
						rf.leader = rf.me
						rf.state = Leader
						rf.l(rf.currentTerm, "win the election\n")
						rf.lock.Unlock()
						win = true
						// 开始向其他follower发送心跳包
						go rf.heartbeat()
						break
					}
					rf.lock.Unlock()
				}
			}
			if win {
				rf.lock_l(fmt.Sprintf("win election votes: %d\n", votes))
			} else {
				rf.lock_l(fmt.Sprintf("fail election votes: %d\n", votes))
			}

		}
	}
}

func (rf *Raft) heartbeat() {
	// 如果自己是leader的话，向其余follower发送心跳包
	for rf.killed() == false {
		is_leader := false
		lock_guard(&rf.lock, func() {
			is_leader = rf.state == Leader
		})
		if is_leader {
			// 构建RPC参数
			rf.lock.Lock()
			rf.l(rf.currentTerm, "sending heartbeats")
			args := AppendEntriesArgs{
				Term:     rf.currentTerm,
				LeaderId: rf.me,
			}
			rf.lock.Unlock()

			// 创建go routine用于发送RPC
			type AppendEntriesReplyRet struct {
				Reply AppendEntriesReply
				Ok    bool
			}
			res_channel := make(chan AppendEntriesReplyRet, len(rf.peers)-1)
			start := time.Now()
			for i := 0; i < len(rf.peers); i++ {
				t := i
				if i != rf.me {
					go func() {
						reply := AppendEntriesReply{}
						if rf.peers[t].Call("Raft.AppendEntries", &args, &reply) {
							res_channel <- AppendEntriesReplyRet{Reply: reply, Ok: true}
						} else {
							res_channel <- AppendEntriesReplyRet{Reply: reply, Ok: false}
						}
					}()
				}
			}

			// 发送完毕，等消息
			for i := 0; i < len(rf.peers)-1; i++ {
				replyRet := <-res_channel
				rf.lock.Lock()
				if rf.state != Leader {
					rf.l(rf.currentTerm, "no longer Leader, stop heartbeats")
					rf.lock.Unlock()
					return
				}
				if replyRet.Ok {
					reply := replyRet.Reply
					rf.l(rf.currentTerm, fmt.Sprintf("beats #%d", i))
					if reply.Term > rf.currentTerm {
						rf.state = Follower
						rf.leader = -1
						rf.currentTerm = reply.Term
						rf.votedFor = none
						rf.lock.Unlock()
						return
					}
				}
				rf.lock.Unlock()
			}
			delta := time.Since(start)
			rf.lock_l(fmt.Sprintf("time in heartbeat: %s", delta))
			time.Sleep(heartbeatInterval)
		}
	}
}

func (rf *Raft) l(term int, msg string) {
	log.Println(fmt.Sprintf("term[%d] #%d%s: %s", term, rf.me, rf.st(), msg))
}

func (rf *Raft) lock_l(msg string) {
	rf.lock.Lock()
	defer rf.lock.Unlock()
	log.Println(fmt.Sprintf("term[%d] #%d%s: %s", rf.currentTerm, rf.me, rf.st(), msg))
}

func (rf *Raft) st() string {
	return fmt.Sprintf("[%d,%d,%d]", rf.currentTerm, rf.state, rf.leader)
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	log.SetFlags(log.Lmicroseconds)
	log.SetOutput(ioutil.Discard)
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.lastHeartBeat = time.UnixMilli(0)
	rf.currentTerm = 0
	rf.state = Follower

	// Your initialization code here (2A, 2B, 2C).
	rf.leader = -1

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	// start ticker goroutine to start elections
	go rf.ticker()

	return rf
}
